package com.demon.framework.boot.failover;

import com.alibaba.fastjson.JSON;
import com.demon.framework.boot.context.SpringContext;
import com.demon.framework.boot.job.AbstractTask;
import com.demon.framework.boot.job.JobForTaskService;
import com.demon.framework.util.serialize.ProtostuffUtil;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.time.DateUtils;
import org.springframework.beans.factory.config.AutowireCapableBeanFactory;
import org.springframework.http.HttpMethod;
import org.springframework.http.RequestEntity;
import org.springframework.http.ResponseEntity;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.support.GeneratedKeyHolder;
import org.springframework.jdbc.support.KeyHolder;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.AsyncResult;
import org.springframework.util.Assert;
import org.springframework.util.Base64Utils;
import org.springframework.web.client.RestTemplate;

import java.net.URI;
import java.net.URISyntaxException;
import java.sql.PreparedStatement;
import java.sql.Statement;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Future;

/**
 * 业务重试 服务实现
 *
 * @author Demon-HY
 * @date 2020/7/24 16:44
 */
@Slf4j
public class TaskFailoverService {

    private final JobForTaskService jobForTaskService;
    private final AutowireCapableBeanFactory capableBeanFactory;
    private final JdbcTemplate jdbcTemplate;
    /**
     * 缓存 重试任务 实例
     */
    private final Map<Class<? extends AbstractTaskFailoverJob>, AbstractTaskFailoverJob>
            cacheFailOverJob = Maps.newConcurrentMap();

    public TaskFailoverService(JobForTaskService jobForTaskService, AutowireCapableBeanFactory capableBeanFactory, JdbcTemplate jdbcTemplate) {
        this.jobForTaskService = jobForTaskService;
        this.capableBeanFactory = capableBeanFactory;
        this.jdbcTemplate = jdbcTemplate;
    }

    /**
     * 添加需要重试的任务  使用异步 与主线程事务隔离
     *
     * @param jobClass      任务处理类class
     * @param requestEntity 任务重放 内容
     * @param reason        加入重试任务的原因
     * @return 返回重试任务的主键ID
     */
    @Async
    public Future<Long> addRetryTask(Class<? extends AbstractTaskFailoverJob> jobClass, RequestEntity requestEntity, String reason) {
        return addRetryTask(jobClass, requestEntity, reason, StringUtils.EMPTY);
    }

    /**
     * 添加需要重试的任务  使用异步 与主线程事务隔离
     *
     * @param jobClass      任务处理类class
     * @param requestEntity 任务重放 内容
     * @param reason        加入重试任务的原因
     * @param taskName      重试任务的名称
     * @return 返回重试任务的主键ID
     */
    @Async
    public Future<Long> addRetryTask(Class<? extends AbstractTaskFailoverJob> jobClass, RequestEntity requestEntity, String reason, String taskName) {

        Assert.notNull(jobClass, "重试的类型不能为空");
        Assert.notNull(requestEntity, "重试的请求不能为空");
        Assert.notNull(reason, "上次失败的理由不能为空");

        AbstractTaskFailoverJob taskFailoverJob = getTaskFailoverJob(jobClass);

        String taskType = taskFailoverJob.getTaskType();

        //如果任务name 不为空，type相同，则清除原task任务
        delRetryTask(taskType, taskName);

        TaskFailoverBean taskFailoverBean = new TaskFailoverBean();

        taskFailoverBean.setUrl(requestEntity.getUrl().toString());
        taskFailoverBean.setData(JSON.toJSONString(requestEntity));
        taskFailoverBean.setName(taskName);
        taskFailoverBean.setSerializeData(ProtostuffUtil.serializeToBase64(requestEntity));
        taskFailoverBean.setReason(reason);


        taskFailoverBean.setType(taskType);

        //首次重试间隔,
        int nextIntervalTime = taskFailoverJob.getNextIntervalTime(0);

        Date nextTime = DateUtils.addSeconds(new Date(), nextIntervalTime);
        taskFailoverBean.setNextTime(nextTime);
        //保存重试任务
        KeyHolder keyHolder = new GeneratedKeyHolder();
        getJdbcTemplate().update(conn -> {
            PreparedStatement ps = conn.prepareStatement(taskFailoverBean.getSqlForAdd(), Statement.RETURN_GENERATED_KEYS);
            return ps;
        }, keyHolder);
        taskFailoverBean.setId(keyHolder.getKey().longValue());


        jobForTaskService.addJob(new AbstractTask(taskFailoverBean.getId().toString()) {
            @Override
            protected Class<? extends AbstractTaskFailoverJob> getJobClass() {
                return jobClass;
            }

            @Override
            protected String getJobGroup() {
                return TaskFailoverBean.JOB_GROUP_FOR_FAILOVER;
            }
        }, nextTime);

        return new AsyncResult<>(taskFailoverBean.getId());
    }


    /**
     * 删除重试任务
     *
     * @param taskId
     */
    @Async
    public void delRetryTask(Long taskId) {
        //清除task
        log.info("清除重试任务,taskId:{}", taskId);
        getJdbcTemplate().execute(TaskFailoverBean.getSqlForClear(taskId));

        //清除job
        jobForTaskService.delJob(taskId.toString(), TaskFailoverBean.JOB_GROUP_FOR_FAILOVER);
    }

    /**
     * 删除重试任务
     *
     * @param taskType 重试任务的分类
     * @param taskName 重试任务的名称
     */
    @Async
    public void delRetryTask(String taskType, String taskName) {
        //如果任务name 不为空，type相同，则清除原task任务
        log.info("清除重试任务,taskType:{},taskName:{}", taskType, taskName);
        if (StringUtils.isNotEmpty(taskName)) {
            try {
                List<TaskFailoverBean> taskWillClearList = getJdbcTemplate().query(TaskFailoverBean.getSqlForFindByTypeAndName(taskType, taskName), TaskFailoverBean.BEAN_BEAN_PROPERTY_ROW_MAPPER);
                if (null != taskWillClearList && !taskWillClearList.isEmpty()) {
                    if (log.isInfoEnabled()) {
                        log.info("接收到相同类型的任务,清除原始的任务:{}", JSON.toJSONString(taskWillClearList));
                    }
                    taskWillClearList.forEach(taskWillClear -> delRetryTask(taskWillClear.getId()));
                }
            } catch (Exception e) {
                log.error("清除相同任务失败,taskType:{},taskName:{}", taskType, taskName, e);
            }

        }
    }

    /**
     * 删除重试任务
     *
     * @param jobClass 重试任务的处理类名
     * @param taskName 重试任务的名称
     */
    @Async
    public void delRetryTask(Class<? extends AbstractTaskFailoverJob> jobClass, String taskName) {
        AbstractTaskFailoverJob taskFailoverJob = getTaskFailoverJob(jobClass);
        delRetryTask(taskFailoverJob.getTaskType(), taskName);
    }

    /**
     * 获取数据库连接
     *
     * @return
     */
    protected JdbcTemplate getJdbcTemplate() {
        return this.jdbcTemplate;
    }

    /**
     * 根据任务处理类获取 其 实类
     *
     * @param jobClass
     * @return
     */
    private AbstractTaskFailoverJob getTaskFailoverJob(Class<? extends AbstractTaskFailoverJob> jobClass) {
        if (!cacheFailOverJob.containsKey(jobClass)) {
            //double check
            synchronized (jobClass) {
                if (!cacheFailOverJob.containsKey(jobClass)) {
                    //先从 Spring容器取 jobClass
                    AbstractTaskFailoverJob taskFailoverJob = SpringContext.getBean(jobClass);
                    if (null == taskFailoverJob) {
                        try {
                            taskFailoverJob = jobClass.newInstance();
                        } catch (Throwable e) {
                            throw new UnsupportedOperationException("创建新对象异常,class:{" + jobClass + "}", e);
                        }
                        capableBeanFactory.autowireBean(taskFailoverJob);
                    }
                    cacheFailOverJob.put(jobClass, taskFailoverJob);
                }
            }

        }

        return cacheFailOverJob.get(jobClass);
    }


    public static void main(String[] args) throws URISyntaxException {

        RequestEntity requestEntity = new RequestEntity("Ss=ss", HttpMethod.POST, new URI("http://www.baidu.com"));
        RestTemplate restTemplate = new RestTemplate();

        System.out.println(JSON.toJSONString(requestEntity));
        long start = System.currentTimeMillis();

        byte[] bytes = ProtostuffUtil.serializeList(Lists.newArrayList(requestEntity));


        System.out.println(bytes);
        System.out.println(Base64Utils.encodeToUrlSafeString(bytes));
        System.out.println(System.currentTimeMillis() - start);


        List<RequestEntity> requestEntities = ProtostuffUtil.deserializeList(bytes, RequestEntity.class);

        System.out.println(System.currentTimeMillis() - start);

        System.out.println(requestEntities.get(0));


        ResponseEntity<String> exchange = restTemplate.exchange(requestEntity, String.class);

//        System.out.println(exchange.getBody());
    }

}
